//package com.yesun.task.pipeline.worker;
//
//import com.sun.org.apache.xpath.internal.functions.FuncTrue;
//import com.yesun.task.pipeline.common.Constant;
//import com.yesun.task.pipeline.common.JobUtil;
//import com.yesun.task.pipeline.domain.dto.JobDto;
//import com.yesun.task.pipeline.domain.dto.TaskDto;
//import com.yesun.task.pipeline.handler.Executor;
//import com.yesun.task.pipeline.handler.Loader;
//import com.yesun.task.pipeline.sink.ClickhouseSink;
//import lombok.extern.slf4j.Slf4j;
//
//import java.util.*;
//import java.util.concurrent.*;
//
//@Slf4j
//public class Worker extends Thread{
//    private ExecutorService executorService = Executors.newFixedThreadPool(4);
//    volatile boolean stop = false;
//    private static LinkedBlockingQueue<TaskDto> queue = new LinkedBlockingQueue<TaskDto>();
//    private static Map<String,Executor> executorMap = new HashMap<>();
//
//    //统计JOB执行情况，保存TaskName和JobName
//    public static List<String> LIST_JOBS = new ArrayList<>();
//    public static Map<String, String> MAP_TASK = new HashMap();
//
//    //注册
//    public static void registerLoader(String name, Loader loader) {
//        log.info("register loader:" + name + " - "+ loader);
//        //执行loader，每次触发就执行
//        loader.load(queue);
//    }
//    public static void registerExecutor(String name, Executor executor) {
//        //重复注册会覆盖，但还是做判断，如果存在就不注册
//        if(!executorMap.containsKey(name)) {
//            log.info("register executor:" + name + " - "+ executor);
//            executorMap.put(name, executor);
//        }
//    }
//
//    @Override
//    public void run() {
//        log.info("Worker staring...");
//        //线程池状态监测
//        Monitor monitor = new Monitor();
//        monitor.setName("ThreadPoolExecutorMoniterThread");
//        monitor.start();
//
//        //启动实时队列
//        ClickhouseSink.BatchInsertWorker batchInsertWorker = new ClickhouseSink.BatchInsertWorker();
//        batchInsertWorker.start();
//
//        while (!stop) {
//            try {
//                final TaskDto task = queue.take();
//                log.info("queue take task:" + task.getName());
//                //处理业务
////                executor.execute(new RunableWithName(){ //传递一个name进去
////                    @Override
////                    public String getName() {
////                        return task.getName(); //需要传递JobName，因为一个JOB可能有N个task
////                    }
////                    @Override
////                    public void run() {
////                        //取得任务，分发给对应的Executor处理
////                        Executor executor = executorMap.get(task.getTag());
////                        if(executor != null) {
////                            log.info("execute task:" + task.getName() + " executor:" + executor.toString());
////                            executor.execute(task);
////                            log.info("execute task finished! " + task.getName());
////                        }
////                    }
////                });
//
//                //为了统计，触发后才加入
//                if(!LIST_JOBS.contains(task.getJobName())) {
//                    LIST_JOBS.add(task.getJobName());
//                }
//                MAP_TASK.put(task.getName(), task.getJobName());//重复的会更新
//
//                if(task.getTag().equals(Constant.JOB_TAG_RT_DETAIL)) {
//                    //实时任务直接启动线程，不走线程池
//                    //取得任务，分发给对应的Executor处理
//                    Executor executor = executorMap.get(task.getTag());
//                    if(executor != null) {
//                        RunableWithName runableWithName = new RunableWithName(){
//                            @Override
//                            public void run() {
//                                log.info("execute task:" + task.getName() + " executor:" + executor.toString());
//                                executor.execute(task);
//                                log.info("execute task finished! " + task.getName());
//                            }
//                            @Override
//                            public String getName() {
//                                return task.getName();
//                            }
//                        };
//                        //直接启动
//                        new Thread(runableWithName).start();
//                    }
//                } else {
//                    //分发给executor执行
//                    Future<String> future = executorService.submit(new CallableWithName<String>(){
//                        //传递一个name进去
//                        @Override
//                        public String getName() {
//                            return task.getName(); //需要传递JobName，因为一个JOB可能有N个task
//                        }
//                        @Override
//                        public String call() {
//                            //取得任务，分发给对应的Executor处理
//                            Executor executor = executorMap.get(task.getTag());
//                            if(executor != null) {
//                                log.info("execute task:" + task.getName() + " executor:" + executor.toString());
//                                executor.execute(task);
//                                log.info("execute task finished! " + task.getName());
//                                return "success";
//                            }
//                            return "fail";
//                        }
//                    });
//                    try {
//                        String result = future.get();
//                        //成功，写入
//                        log.info("task["+task.getName()+"] 执行完毕，返回结果：" + result);
//                        //从TASK_MAP清理
//                        MAP_TASK.remove(task.getName());
//                    } catch (ExecutionException e) {
//                        e.printStackTrace();
//                    }
//                }
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
//    }
//
//    public class Monitor extends Thread {
//        @Override
//        public void run() {
//            log.info("Monitor staring...");
//            while (true) {
//                try {
//                    Thread.sleep(30000);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//
//                ThreadPoolExecutor threadPoolExecutor = ((ThreadPoolExecutor) executorService);
//                int threadCount = threadPoolExecutor.getActiveCount();
//                long taskCount = threadPoolExecutor.getTaskCount();
//                long completeTaskCount = threadPoolExecutor.getCompletedTaskCount();
//                log.info("****************************");
//                log.info("******活动线程数：" + threadCount);
//                log.info("******全部任务数：" + taskCount);
//                log.info("******完成任务数：" + completeTaskCount);
//                log.info("****************************");
//            }
//        }
//    }
//}
